package cc.sunwl.study.java8.异步编程;

import cc.sunwl.study.java8.异步编程.domain.Discount;
import cc.sunwl.study.java8.异步编程.domain.Quote;
import cc.sunwl.study.java8.异步编程.domain.Shop;

import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class FutureTest {
    private static List<Shop> shops = null;

    static {
        shops = Arrays.asList(
                new Shop("BestPrice"),
                new Shop("LetsSaveBig"),
                new Shop("MyFavoriteShop"),
                new Shop("BuyItAll")
        );
    }

    public static void main(String[] args) {
        Shop shop = new Shop("BestShop");
        long start0 = System.nanoTime();
        // 查询商店，试图取得商品的价格
        Future<Double> futurePrice = shop.getPriceAsync("my favorite product");
        long invocationTime = ((System.nanoTime() - start0) / 1_000_000);
        System.out.println("Invocation returned after " + invocationTime + " msecs");

        // 执行更多任务，比如查询其他商店
        doSomeThingElse();
        // 在计算商品价格的同时
        try {
            // 在Future对象读取价格，如果价格未知，会发生阻塞
            double price = futurePrice.get();
            System.out.printf("price is %.2f%n", price);

        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        long retrievalTime = ((System.nanoTime() - start0) / 1_000_000);
        System.out.println("Price returned after " + retrievalTime + " msecs");

        System.out.println("----------------------------------");


        long start1 = System.nanoTime();
        List<String> prices1 = findPrices("iphone7");
        System.out.println("prices1 = " + prices1);
        long duration1 = (System.nanoTime() - start1) / 1_000_000;
        System.out.println("Done in " + duration1 + " msecs");

        System.out.println("----------------------------------");

        long start2 = System.nanoTime();
        List<String> prices2 = findPricesV2("iphone7");
        System.out.println("prices2 = " + prices2);
        long duration2 = (System.nanoTime() - start2) / 1_000_000;
        System.out.println("Done in " + duration2 + " msecs");

        System.out.println("----------------------------------");

        long start3 = System.nanoTime();
        List<String> prices3 = findPricesV3("iphone7");
        System.out.println("prices3 = " + prices3);
        long duration3 = (System.nanoTime() - start3) / 1_000_000;
        System.out.println("Done in " + duration3 + " msecs");

        System.out.println("----------------------------------");

        long start4 = System.nanoTime();
        List<String> prices4 = findPricesV4("iphone7");
        System.out.println("prices4 = " + prices4);
        long duration4 = (System.nanoTime() - start4) / 1_000_000;
        System.out.println("Done in " + duration4 + " msecs");

        System.out.println("查询商店，并带有折扣信息");
        System.out.println("----------------------------------");

        long startByDiscount = System.nanoTime();
        List<String> pricesByDiscount = findPricesByDiscount("iphone7");
        System.out.println("pricesByDiscount = " + pricesByDiscount);
        long durationByDiscount = (System.nanoTime() - startByDiscount) / 1_000_000;
        System.out.println("Done in " + durationByDiscount + " msecs");

        System.out.println("----------------------------------");

        long startByDiscount2 = System.nanoTime();
        List<String> pricesByDiscount2 = findPricesByDiscountV2("iphone7");
        System.out.println("pricesByDiscount2 = " + pricesByDiscount2);
        long durationByDiscount2 = (System.nanoTime() - startByDiscount2) / 1_000_000;
        System.out.println("Done in " + durationByDiscount2 + " msecs");

        System.out.println("----------------------------------");

        long start = System.nanoTime();
        CompletableFuture[] futures = findPricesStream("iphone X")
                .map(f -> f.thenAccept(
                        s -> System.out.println(s + " (done in " + ((System.nanoTime() - start) / 1_000_000) + " msecs)")
                )).toArray(size -> new CompletableFuture[size]);
        CompletableFuture.allOf(futures).join();
        System.out.println("All shops have now responded in " + ((System.nanoTime() - start) / 1_000_000) + " msecs");
    }

    /**
     * 定制执行器<br>
     * 取商店与100之间最小值作为线程数
     */
    private final static Executor executor = Executors.newFixedThreadPool(Math.min(shops.size(), 100), (r) -> {
        Thread t = new Thread(r);
        t.setDaemon(true);
        return t;
    });

    private static void doSomeThingElse() {
        System.out.println("FutureTest.doSomeThingElse");
    }

    /**
     * 查询指定商品在指定商店中的价格（同步）<br>
     * 当前线程为：main<br>
     * 当前线程为：main<br>
     * 当前线程为：main<br>
     * 当前线程为：main<br>
     * prices1 = [BestPrice price is 119.92, LetsSaveBig price is 215.99, MyFavoriteShop price is 121.89, BuyItAll price is 153.56]<br>
     * Done in 4010 msecs，因为对这四个商店是顺序进行的
     *
     * @param product 商品名称
     * @return 价格列表
     */
    private static List<String> findPrices(String product) {
        return shops.stream().map(shop -> String.format("%s price is %.2f", shop.getName(), shop.getPrice(product))).collect(Collectors.toList());
    }

    /**
     * 查询指定商品在指定商店中的价格（同步）<br>
     * 并行流实现是ForkJoinPool<br>
     * 当前线程为：main<br>
     * 当前线程为：ForkJoinPool.commonPool-worker-1<br>
     * 当前线程为：main<br>
     * 当前线程为：ForkJoinPool.commonPool-worker-1<br>
     * prices2 = [BestPrice price is 139.80, LetsSaveBig price is 134.01, MyFavoriteShop price is 138.93, BuyItAll price is 193.43]<br>
     * Done in 2002 msecs
     *
     * @param product 商品名称
     * @return 价格列表
     */
    private static List<String> findPricesV2(String product) {
        System.out.println("可运行的处理器个数（并行流）：" + Runtime.getRuntime().availableProcessors());
        return shops.parallelStream().map(shop -> String.format("%s price is %.2f", shop.getName(), shop.getPrice(product))).collect(Collectors.toList());
    }

    /**
     * 查询指定商品在指定商店中的价格（异步）<br>
     * 当前线程为：Thread-2<br>
     * 当前线程为：Thread-4<br>
     * 当前线程为：Thread-1<br>
     * 当前线程为：Thread-3<br>
     * prices3 = [BestPrice price is 150.82876356192423, LetsSaveBig price is 186.44306741118828, MyFavoriteShop price is 169.30840164577876, BuyItAll price is 144.35567841565353]<br>
     * Done in 1005 msecs
     *
     * @param product 商品名称
     * @return 价格列表
     */
    private static List<String> findPricesV3(String product) {
        // return shops.stream().map(shop -> CompletableFuture.supplyAsync(()->shop.getName() + " price is " + shop.getPrice(product))).collect(Collectors.toList());
        // 使用CompletableFuture以异步方式计算每种商品的价格
        List<CompletableFuture<String>> priceFutures = shops.stream().map(shop -> CompletableFuture.supplyAsync(() -> shop.getName() + " price is " + shop.getPrice(product))).collect(Collectors.toList());
        // CompletableFuture类中的join与Future接口中的get有相同的含义。唯一不同的是，join不会抛出任何检测到的异常
        // 等待所有异步操作结束
        return priceFutures.stream().map(CompletableFuture::join).collect(Collectors.toList());
    }

    /**
     * 同V3版本，只是采用定制执行器
     *
     * @param product 商品名称
     * @return 价格列表
     */
    private static List<String> findPricesV4(String product) {
        // return shops.stream().map(shop -> CompletableFuture.supplyAsync(()->shop.getName() + " price is " + shop.getPrice(product))).collect(Collectors.toList());
        // 使用CompletableFuture以异步方式计算每种商品的价格
        List<CompletableFuture<String>> priceFutures = shops.stream()
                .map(shop -> CompletableFuture.supplyAsync(() -> shop.getName() + " price is " + shop.getPrice(product), executor))
                .collect(Collectors.toList());
        // CompletableFuture类中的join与Future接口中的get有相同的含义。唯一不同的是，join不会抛出任何检测到的异常
        // 等待所有异步操作结束
        return priceFutures.stream()
                .map(CompletableFuture::join)
                .collect(Collectors.toList());
    }

    private static List<String> findPricesByDiscount(String product) {
        return shops.stream()
                .map(shop -> shop.getPriceV2(product))
                .map(Quote::parse)
                .map(Discount::applyDiscount)
                .collect(Collectors.toList());
    }

    private static List<String> findPricesByDiscountV2(String product) {
        List<CompletableFuture<String>> priceFutures = shops.stream()
                // 以异步方式取得每个shop中指定产品的原始价格。...Async方法默认使用ForkJoinPool.commonPool()，或者使用重载方法，指定一个
                .map(shop -> CompletableFuture.supplyAsync(() -> shop.getPriceV2(product), executor))
                // Quote对象存在时，对其返回的值进行转换
                .map(future -> future.thenApply(Quote::parse))
                // 使用另一个异步任务构造期望的Future，申请折扣
                .map(future -> future.thenCompose(quote -> CompletableFuture.supplyAsync(() -> Discount.applyDiscount(quote), executor)))
                .collect(Collectors.toList());
        return priceFutures.stream()
                // 等待流中的所有Future执行完毕，并提取各自的返回值
                .map(CompletableFuture::join)
                .collect(Collectors.toList());
    }

    /**
     *
     * @param product 产品名称
     * @return Stream<CompletableFuture<String>>
     */
    private static Stream<CompletableFuture<String>> findPricesStream(String product) {
        return shops.stream()
                .map(shop -> CompletableFuture.supplyAsync(
                        () -> shop.getPriceV2(product), executor
                ))
                .map(future -> future.thenApply(Quote::parse))
                .map(future -> future.thenCompose(
                        quote -> CompletableFuture.supplyAsync(
                                () -> Discount.applyDiscountV2(quote), executor
                        )
                ));
    }

    /**
     * 题外话<br>
     * 两个互不相关的Future合并，第一个supplyAsync作为thenCombine的参数
     */
    private void testCombine() {
        CompletableFuture<Integer> vCompletableFuture = CompletableFuture
                .supplyAsync(
                        () -> new Random().nextInt(1000)
                )
                .thenCombine(
                        CompletableFuture.supplyAsync(
                                () -> new Random().nextInt(2000)
                        ),
                        (price, rate) -> price * rate
                );
    }
}
